package com.google.code.p.restrpc.directory;

import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;

import com.google.code.p.restrpc.client.RpcProxy;
import com.google.code.p.restrpc.client.proxy.RpcProxyFactory;
import com.google.code.p.restrpc.core.RpcServiceException;
import com.google.code.p.restrpc.event.RpcEventObject;
import com.google.code.p.restrpc.service.NotifyService;

/**
 * @author dz
 */
public class DefaultRegistryListener implements RegistryEventListener {
    
    private int delayOfRetry = 60;
    
    private Registry registryService;
    
    private RpcProxy rpcProxy = RpcProxyFactory.getInstance();
    
    private ExecutorService executorService = Executors.newFixedThreadPool(Runtime.getRuntime()
                                                                                  .availableProcessors());
    
    private ScheduledExecutorService scheduledExecutorService = Executors.newScheduledThreadPool(Runtime.getRuntime()
                                                                                                        .availableProcessors());
    
    public DefaultRegistryListener(Registry registryService) {
        this.registryService = registryService;
    }
    
    public int getDelayOfRetry() {
        return delayOfRetry;
    }
    
    public void setDelayOfRetry(int delayOrRetry) {
        if (delayOrRetry < 0) {
            throw new IllegalArgumentException("delayOrRetry can't less than zero!");
        }
        this.delayOfRetry = delayOrRetry;
    }
    
    public boolean support(RpcEventObject event) {
        return ((event instanceof ServiceDisabledEventObject) || (event instanceof ServiceEnabledEventObject));
    }
    
    public void handle(RpcEventObject event) {
        if (event instanceof ServiceDisabledEventObject) {
            onServiceDisabled((ServiceDisabledEventObject) event);
        }
        else if (event instanceof ServiceEnabledEventObject) {
            onServiceEnabled((ServiceEnabledEventObject) event);
        }
    }
    
    protected void onServiceEnabled(ServiceEnabledEventObject eventObject) {
        doServiceEnabled((RemoteServiceDefinition) eventObject.getSource());
    }
    
    protected void onServiceDisabled(ServiceDisabledEventObject eventObject) {
        doServiceDisabled((RemoteServiceDefinition) eventObject.getSource());
    }
    
    protected void doServiceEnabled(final RemoteServiceDefinition serviceDefinition) {
        final NotifyService notifyServcie = rpcProxy.createProxy(NotifyService.class,
                                                                     serviceDefinition.toFullUrl());
        executorService.execute(new Runnable() {
            
            public void run() {
                try {
                    notifyServcie.enable(serviceDefinition.getRemoteAddress(),
                                         serviceDefinition.getRemotePort());
                }
                catch (RpcServiceException e) {
                    throw e;
                }
                
                retryEnableService(notifyServcie, serviceDefinition, 60);
            }
            
        });
    }
    
    protected void doServiceDisabled(final RemoteServiceDefinition serviceDefinition) {
        final NotifyService notifyServcie = rpcProxy.createProxy(NotifyService.class,
                                                                     serviceDefinition.toFullUrl());
        executorService.execute(new Runnable() {
            
            public void run() {
                try {
                    notifyServcie.disable(serviceDefinition.getRemoteAddress(),
                                          serviceDefinition.getRemotePort());
                }
                catch (RpcServiceException e) {
                    throw e;
                }
                
                retryDisableService(notifyServcie, serviceDefinition, 60);
            }
            
        });
    }
    
    public void retryEnableService(final NotifyService notifyServcie,
                                   final RemoteServiceDefinition serviceDefinition,
                                   int delay) {
        scheduledExecutorService.schedule(new Runnable() {
            
            public void run() {
                try {
                    notifyServcie.disable(serviceDefinition.getRemoteAddress(),
                                          serviceDefinition.getRemotePort());
                }
                catch (RpcServiceException e) {
                    throw e;
                }
                
                retryDisableService(notifyServcie, serviceDefinition, 60);
            }
            
        }, delay, TimeUnit.SECONDS);
    }
    
    public void retryDisableService(final NotifyService notifyServcie,
                                    final RemoteServiceDefinition serviceDefinition,
                                    int delay) {
        scheduledExecutorService.schedule(new Runnable() {
            
            public void run() {
                try {
                    notifyServcie.disable(serviceDefinition.getRemoteAddress(),
                                          serviceDefinition.getRemotePort());
                }
                catch (RpcServiceException e) {
                    throw e;
                }
                
                retryDisableService(notifyServcie, serviceDefinition, 60);
            }
            
        }, delay, TimeUnit.SECONDS);
    }
    
}
